package com.xx.rpc.core.remoting.invoker;

import com.xx.rpc.core.registry.Register;
import com.xx.rpc.core.registry.impl.LocalRegister;
import com.xx.rpc.core.remoting.net.params.BaseCallback;
import com.xx.rpc.core.remoting.net.params.XxRpcFutureResponse;
import com.xx.rpc.core.remoting.net.params.XxRpcResponse;
import com.xx.rpc.core.util.XxRpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * rpc消费者工厂
 */
public class XxRpcInvokerFactory {
    private static final Logger logger = LoggerFactory.getLogger(XxRpcInvokerFactory.class);

    // ---------------------- 默认实例 ----------------------

    private static volatile XxRpcInvokerFactory instance = new XxRpcInvokerFactory(LocalRegister.class, null);

    public static XxRpcInvokerFactory getInstance() {
        return instance;
    }

    // ---------------------- config ----------------------

    //服务登记类
    private Class<? extends Register> serviceRegistryClass;
    //服务登记参数
    private Map<String, String> serviceRegistryParam;
    //服务登录实现对象
    private Register register;

    public Register getRegister() {
        return register;
    }

    //响应回调列表
    private List<BaseCallback> stopCallbackList = new ArrayList<BaseCallback>();

    public void addStopCallBack(BaseCallback callback) {
        stopCallbackList.add(callback);
    }

    public XxRpcInvokerFactory() {
    }

    public XxRpcInvokerFactory(Class<? extends Register> serviceRegistryClass, Map<String, String> serviceRegistryParam) {
        this.serviceRegistryClass = serviceRegistryClass;
        this.serviceRegistryParam = serviceRegistryParam;
    }

    /**
     * 开启工厂
     *
     * @throws Exception 错误信息
     */
    public void start() throws Exception {
        if (serviceRegistryClass != null) {
            //实例对象
            register = serviceRegistryClass.newInstance();
            //传入登记的服务参数对象启动
            register.start(serviceRegistryParam);
        }
    }

    /**
     * 停止服务
     *
     * @throws Exception 错误信息
     */
    public void stop() throws Exception {
        // stop registry
        if (register != null) {
            register.stop();
        }

        // 处理完响应回调
        if (stopCallbackList.size() > 0) {
            for (BaseCallback callback : stopCallbackList) {
                try {
                    callback.run();
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }

        // 暂停响应回调线程池
        stopCallbackThreadPool();
    }
// ---------------------- 响应回调线程池 ----------------------

    private ThreadPoolExecutor responseCallbackThreadPool = null;

    public void executeResponseCallback(Runnable runnable) {

        if (responseCallbackThreadPool == null) {
            synchronized (this) {
                if (responseCallbackThreadPool == null) {
                    responseCallbackThreadPool = new ThreadPoolExecutor(
                            10,
                            100,
                            60L,
                            TimeUnit.SECONDS,
                            new LinkedBlockingQueue<Runnable>(1000),
                            new ThreadFactory() {
                                @Override
                                public Thread newThread(Runnable r) {
                                    return new Thread(r, "xx-rpc, 响应回调线程池执行-" + r.hashCode());
                                }
                            },
                            new RejectedExecutionHandler() {
                                @Override
                                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                                    throw new XxRpcException("xx-rpc 响应回调线程池执行报错!");
                                }
                            });        // default maxThreads 300, minThreads 60
                }
            }
        }
        responseCallbackThreadPool.execute(runnable);
    }

    public void stopCallbackThreadPool() {
        if (responseCallbackThreadPool != null) {
            responseCallbackThreadPool.shutdown();
        }
    }

    // ---------------------- 异步处理响应池 ----------------------

    private ConcurrentMap<String, XxRpcFutureResponse> futureResponsePool = new ConcurrentHashMap<String, XxRpcFutureResponse>();
    public void setInvokerFuture(String requestId, XxRpcFutureResponse futureResponse){
        futureResponsePool.put(requestId, futureResponse);
    }
    public void removeInvokerFuture(String requestId){
        futureResponsePool.remove(requestId);
    }
    public void notifyInvokerFuture(String requestId, final XxRpcResponse xxlRpcResponse){

        // get
        final XxRpcFutureResponse futureResponse = futureResponsePool.get(requestId);
        if (futureResponse == null) {
            return;
        }

        // notify
        if (futureResponse.getInvokeCallback()!=null) {

            // callback type
            try {
                executeResponseCallback(new Runnable() {
                    @Override
                    public void run() {
                        if (xxlRpcResponse.getErrorMsg() != null) {
                            futureResponse.getInvokeCallback().onFailure(new XxRpcException(xxlRpcResponse.getErrorMsg()));
                        } else {
                            futureResponse.getInvokeCallback().onSuccess(xxlRpcResponse.getResult());
                        }
                    }
                });
            }catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } else {

            // other nomal type
            futureResponse.setResponse(xxlRpcResponse);
        }

        // do remove
        futureResponsePool.remove(requestId);

    }

}
